package com.gitee.wsl.coroutines.channels.receive

/*
 * Copyright (C) 2022 Square, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
//package app.cash.turbine

import kotlinx.coroutines.channels.Channel
import kotlin.time.Duration
import kotlinx.coroutines.channels.ChannelResult
import kotlinx.coroutines.channels.ClosedReceiveChannelException
import kotlinx.coroutines.channels.ReceiveChannel
import kotlinx.coroutines.channels.onClosed
import kotlinx.coroutines.channels.onFailure
import kotlinx.coroutines.channels.onSuccess
import kotlinx.coroutines.currentCoroutineContext
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flow
import kotlin.coroutines.cancellation.CancellationException
import kotlin.time.Duration.Companion.milliseconds


/**
 * Returns the most recent item that has already been received.
 * If channel was closed with no item being received
 * previously, this function will throw an [AssertionError]. If channel
 * was closed with an exception, this function will throw the underlying exception.
 *
 * @throws AssertionError if no item was emitted.
 */
fun <T> ReceiveChannel<T>.expectMostRecentItem(name: String? = null): T {
    var previous: ChannelResult<T>? = null
    while (true) {
        val current = tryReceive()
        current.exceptionOrNull()?.let { throw it }
        if (current.isFailure) {
            break
        }
        previous = current
    }

    if (previous?.isSuccess == true) return previous.getOrThrow()

    throw AssertionError("No item was found".qualifiedBy(name))
}



private val DEFAULT_TIMEOUT: Duration = 3000.milliseconds

internal suspend fun contextTimeout(): Duration {
    return currentCoroutineContext()[TurbineTimeoutElement]?.timeout ?: DEFAULT_TIMEOUT
}


suspend fun <T> Iterable<T>.toReceiveChannel(capacity: Int = Channel.UNLIMITED): ReceiveChannel<T> =
    Channel<T>(capacity = capacity).apply {
        this@toReceiveChannel.forEach { send(it) }
        close()
    }

 suspend fun <T> Collection<T>.toReceiveChannel(): ReceiveChannel<T> = toReceiveChannel(capacity = size)

